package producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author 夏焰波
 * @date 2020-09-13 19:29
 */
public class MyProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        /**
         * KafkaProducer：需要创建一个生产者对象，用来发送数据
         * ProducerConfig：获取所需的一系列配置参数
         * ProducerRecord：每条数据都要封装成一个ProducerRecord 对象
         */
        //1.创建Kafka生产者的配置信息
        Properties props = new Properties();
        //2.kafka 集群， broker-list
        props.put("bootstrap.servers", "hadoop102:9092");
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");

        //3.ACK应答级别
        props.put("acks", "all");
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        //4.重试次数
        props.put("retries", 4);
        //5.批次大小 16k
        props.put("batch.size", 16384);
        //6.等待时间
        props.put("linger.ms", 1);
        //7.RecordAccumulator 缓冲区大小 32M
        props.put("buffer.memory", 33554432);
        //8.Key，Value的序列化类
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        //9.创建生成者对象
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            //10.发送数据
            Future<RecordMetadata> future = producer.send(new ProducerRecord<String, String>("first",
                    "key-" + Integer.toString(i), Integer.toString(i)));

            /**
             * 同步发送的意思就是，一条消息发送之后，会阻塞当前线程， 直至返回 ack。
             * 由于send方法返回的是一个 Future对象，根据 Futrue对象 的特点，我们也可以实现 同
             * 步发送的效果 ，只需在调用 Future对象的 get方发即可。
             */
            RecordMetadata metadata = future.get();

            //11.关闭资源，会清缓存
            producer.close();

        }
    }
}
